#include <queue>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <jobQueue.h>
#include <type.h>
#include <taskManager.h>
#include <modules.h>
#include <cnotify.h>
#include <framework.h>
#include <commands.h>
#include <keyMap.h>

class WorkQ : public objbase
{
private:
    using VoidEntryFunc = std::function<void(void)>;
    DISABLE_COPY_CLASS(WorkQ);
    std::queue<VoidEntryFunc> tasks;
    std::mutex m_lock;
    std::condition_variable cv_task;
    std::atomic<bool> stoped;

public:
    DECLARE_CTOR_DTOR(WorkQ);

    STATUS preInit()
    {
        return OK;
    }

    STATUS postInit()
    {
        auto lambda = [](void* pobjPara) -> int
            {
                WorkQ* pThis = static_cast<WorkQ*>(pobjPara);
                pThis->Proc();
                return 0;
            };

        using FuncType = int (*)(void*);
        FuncType funcPtr = lambda;

        COND_CHECK_RET_ERROR(CREATE_OBJTASK("WorkQueue", reinterpret_cast<FUNCPTR>(funcPtr), this) != OK);

        COND_CHECK_RET_ERROR(registerMessageListener(MODULE_WORKQ_MGR) != OK);

        return OK;
    }

    void Proc()
    {
        while (!this->stoped)
        {
            VoidEntryFunc task;
            {
                std::unique_lock<std::mutex> lock{ this->m_lock };

                this->cv_task.wait(lock,
                    [this]
                    {
                        return this->stoped.load() ||
                            !this->tasks.empty();
                    });
                if (this->stoped && this->tasks.empty())
                {
                    return;
                }
                task = std::move(this->tasks.front());
                this->tasks.pop();
            }
            task();
        }
    }

    STATUS dump(TextTable& t)
    {
        t.setTableName("jobWorkQueue");
        t.add("size");
        t.add("status").endOfRow();

        t.add(std::to_string(tasks.size()));
        t.add(status_to_string(!stoped.load())).endOfRow();

        return OK;
    }

    PROCESS_DESC(process)
    {
        PROCESS_BEGIN()
        PROCESS_CALL(CMD_SYS_SHUTDOWN, sysShutdown)
        PROCESS_END(RET_ERROR)
    }

    STATUS destory()
    {
        stoped.store(true);
        cv_task.notify_all();
        return OK;
    }

    PROCESS_DESC(sysShutdown)
    {
        return destory();
    }
    STATUS commit(VOIDFUNCPTR _func, void* arg)
    {
        COND_CHECK_RET_ERROR(_func == nullptr);
        COND_CHECK_RET_ERROR(stoped.load());

        auto job = [=](){ _func (arg);};

        std::lock_guard<std::mutex> lock{ m_lock };
        tasks.emplace(job);
        cv_task.notify_one();
        
        return OK;
    }
};

REG_TO_FRAMEWORK(WorkQ, MODULE_WORKQ_MGR)

EXTERN_C STATUS jobAdd(VOIDFUNCPTR _func, void * arg)
{
    COND_CHECK_RET_ERROR(_func == nullptr);

    return GET_FRAMEWORK_MGR->callObjectMethod<WorkQ, STATUS>(
        MODULE_WORKQ_MGR, &WorkQ::commit, 
        static_cast<VOIDFUNCPTR>(_func), static_cast<void*>(arg));
}